package com.niit.sparkstreaming

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.util.HashMap

/**
 * @author gaomj
 * @data 2022/11/30 15:29
 */
object LogZaiji {
  def main(args: Array[String]): Unit = {


    // 逻辑：
    // spark steaming :  spark conf  加载配置 local,appName ； streaming context  规定 每隔5s处理数据  ssc
    //                   ssc 读取 kafka 数据
    //               案例 1)   kafka 配置项，  处理kafka 数据  spark streaming ，打印
    //               案例 2) 结果 重新推送到 kafka topic
    //                   什么时候关闭  打开 接收器，接收器关闭之后，关闭。
    // 消费者/消费者组的id
    val group = "niit07"
    // topic 名字
    val topic = "students"

    val conf = new SparkConf().setAppName("bigdataPro").setMaster("local[*]")
    //    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(conf, Seconds(5))


    // 打印日志级别 为 error
    ssc.sparkContext.setLogLevel("error")

    // checkpoint  记录消费的位置信息
    //    ssc.checkpoint("./checkpoint")

    // 配置 kafka 配置项 map 添加 broker 服务，key value 序列化， groupid, 消费方式 earliest ,提交offset
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "niit01:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> group,
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    // 需要scala array 数组
    val topics = Array(topic)

    // 订阅 kafka topic
    val streamRdd = KafkaUtils.createDirectStream(
      ssc,
      PreferConsistent, // alt + 回车键
      Subscribe[String, String](topics, kafkaParams)
    )

    //    streamRdd.foreachRDD(kafkaRDD =>{
    //
    //      if(!kafkaRDD.isEmpty()){
    //        // isEmpty   true 进入
    //        // ！isEmpty    false 进入
    //        // kafka 当中 key 分区号，value 才是你的数据
    //        val lines = kafkaRDD.map(_.value())
    //        lines.foreach(println)
    //
    //      }
    //  }

    // 案例2：
    // 尝试 推送结果到 kafka topic result17
    val lines = streamRdd.map(_.value())
    // 添加 kafka 的producer 配置项  broker 服务  ，key value 序列化
    // 创建 producer
    // 发送数据

    // (word,1) reduceBykey

    val linesRDD = lines.flatMap(_.split(",")).map((_, 1))
      .reduceByKeyAndWindow(
        _ + _, Seconds(10), Seconds(10)
      )

    linesRDD.foreachRDD(x => {


      // kafkaParams  kfkparams
      val kfkparams = new HashMap[String, Object]()
      kfkparams.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "niit01:9092")
      kfkparams.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
      kfkparams.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

      val producer = new KafkaProducer[String, String](kfkparams)

      // 逻辑：
      // 6 个字段
      //         ID,
      //        Name,
      //        Sex,
      //        num,
      //
      //        Status,
      //        Terms
      // 按逗号分隔，并收集
      //      val word = x.flatMap(_.split(",")).collect()
      val word = x.collect()
      // 200005,5    name ,20   1,10  ...

      word.foreach(
        word => {
          if (word._1.equals("L") || word._1.equals("A")) {


            println("=========")
            println(word._1) // 打印状态的那一列
            println("=========")
            // 将 状态 转化为 (状态，1)  =》  (L,1) (A,1)
            // 再用 parallelize 转化为rdd ，因为 reduceByKey 只对rdd 起作用
            //        val rs = word(2).map((_,1))
            val rs = Array(word)
            //      val sc = new SparkContext(conf)
            //      val sex = sc.parallelize(rs)
            val status = ssc.sparkContext.parallelize(rs)
            // reduceByKey 累加  (0,1)  (1,2)
            //      val sexResult = sex.reduceByKey(_+_)
            // 遍历 并发送
            val res = status.foreach(word => {
              println("word=======" + word)
              val status = word._1.toString
              val count = word._2.toString
              val result = status + "," + count
              println("result=======" + result)

            })
            val result = status.collect()
            result.foreach(x => {
              //              println(x)
              val status = x._1.toString
              val count = x._2.toString
              val result = status + "," + count
              producer.send(new ProducerRecord[String, String]("E", result.toString))
            })
          }
        }
      )
    })

    // 什么时候关闭 ssc
    ssc.start()
    ssc.awaitTermination()


  }

}
